package lab3;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import lab3.GraphBuilder.Map;
import lab3.GraphBuilder.Reduce;

import org.apache.hadoop.mapreduce.Reducer;

public class PageRankIter {
	public static class IterMap extends Mapper<LongWritable,Text,Text,Text>{	
		public void map(LongWritable key,Text value,Context context)
				throws  IOException, InterruptedException{
			String line = value.toString();
			// pagekey + \t +pr + \t + pr_link
			String [] tuple = line.split("\t");
			
			double pr_total = Double.parseDouble(tuple[1]);

			if(tuple.length >2) {
				// 有链出的网页
				String[]pages = tuple[2].split(",");
				// 求均值
				double pr = pr_total/pages.length;
				// 转字符串
				String pr_str = String.valueOf(pr);
				// 输出每个页面分的值
				for(String p:pages) {
					context.write(new Text(p),new Text(pr_str));
				}
				// 将pr_link 传到 reduce  "+"作为标记
				context.write(new Text(tuple[0]),new Text("+" + tuple[2]));
			}
		}
	}
	
	public static class IterReduce extends Reducer<Text,Text,Text,Text>{
		public void reduce(Text key, Iterable<Text> values, Context context) 
				throws  IOException, InterruptedException{
			String pr_link = new String();
			double pr = 0; // 计算获得的pr值
			for(Text v:values) {
				String v_str = v.toString();
				if(v_str.startsWith("+")) {
					// 接受pr_link 
					pr_link = v_str.substring(1);//剔除 + 
				}else {
					// 接受到 pr 值
					double tmp = Double.parseDouble(v_str);
					pr += tmp;
				}
			}
			double d = 0.85;
			pr = d*pr + 1 - d;
			String pr_str = String.valueOf(pr);
			
			// 剔除 原先的 pr，替换新的pr
			String new_value = pr_str + "\t" + pr_link;
			
			context.write(key,new Text(new_value));
		}
	}
	public static void main(String[] args)throws Exception {
		// TODO Auto-generated method stub
		
		  Configuration conf = new Configuration();
		  Path path =new Path(args[1]);
	      //加载配置文件 
	      FileSystem fileSystem = path.getFileSystem(conf); 
	      //输出目录若存在则删除 
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
		  Job job = new Job(conf, "PageRank Iter");
		  job.setJarByClass(PageRankIter.class);
		  job.setMapperClass(IterMap.class);
		  job.setReducerClass(IterReduce.class);
		  job.setOutputKeyClass(Text.class);
		  job.setOutputValueClass(Text.class);
		  FileInputFormat.addInputPath(job, new Path(args[0]));
		  FileOutputFormat.setOutputPath(job, new Path(args[1]));
		 job.waitForCompletion(true);
	}

}
